package handlers

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/aliyun/fc-runtime-go-sdk/fccontext"
	"github.com/go-redis/redis/v8"
	"log"
	"net/http"
	"ocan_data_sync_task/config"
	"ocan_data_sync_task/handlers/processor"
	"ocan_data_sync_task/models"
	"ocan_data_sync_task/utils"
	"os"
	"sync"
	"time"
)

// TaskHandler 任务处理器
type TaskHandler struct {
	rdb        *redis.Client
	config     *config.Config
	processors map[models.TaskType]processor.Processor
}

// NewTaskHandler 创建新的任务处理器
func NewTaskHandler(rdb *redis.Client, config *config.Config) *TaskHandler {
	h := &TaskHandler{
		rdb:        rdb,
		config:     config,
		processors: make(map[models.TaskType]processor.Processor),
	}
	h.registerProcessors(
		processor.NewCustomReportProcessor(),
		// processor.NewPromotionProcessor(),
	)
	return h
}

// registerProcessors 注册处理器
func (h *TaskHandler) registerProcessors(processors ...processor.Processor) {
	for _, p := range processors {
		h.processors[p.GetType()] = p
	}
}

// getWorkerCount 获取工作协程数
func (h *TaskHandler) getWorkerCount(workers int) int {
	if workers <= 0 || workers > h.config.Task.MaxWorkers {
		return h.config.Task.DefaultWorkers
	}
	return workers
}

// HttpHandler HTTP 触发器执行函数
func (h *TaskHandler) HttpHandler(ctx context.Context, event *models.HTTPEvent) (interface{}, error) {

	task := models.Task{}
	err := json.Unmarshal([]byte(event.Body), &task)
	if err != nil {
		fmt.Printf("解析body出错: %v\n", err)
	}

	//打印原始数据（格式化的 JSON）
	event.Task = task
	event.Body = ""
	if eventJSON, err := json.MarshalIndent(event, "", "    "); err == nil {
		utils.EchoLog(ctx, "info", fmt.Sprintf("event info:\n%s\n", string(eventJSON)))
	}
	return h.Handle(ctx, &task)
}

// Handle 处理任务
func (h *TaskHandler) Handle(ctx context.Context, task *models.Task) (interface{}, error) {
	if os.Getenv("FC_FUNCTION_NAME") != "" {
		fCtx, _ := fccontext.FromContext(ctx)
		task.ID = fCtx.RequestID
	}

	httpResponse := models.HTTPResponse{
		Code:   http.StatusNotImplemented,
		TaskId: task.ID,
	}

	// 1. 获取对应的处理器
	p, ok := h.processors[task.Type]
	if !ok {
		httpResponse.Message = fmt.Sprintf("未找到任务类型[%s]对应的处理器", task.Type)
		return httpResponse, fmt.Errorf(httpResponse.Message)
	}

	// 2. 参数校验
	if err := p.ValidateRequest(task); err != nil {
		return httpResponse, err
	}

	// 3. 生成任务ID并初始化状态
	taskResult := &models.TaskResult{
		ID:         task.ID,
		Status:     "running",
		Results:    make(map[string]models.ItemTaskResult),
		StartTime:  time.Now().Unix(),
		TotalCount: len(task.Items),
	}

	// 4. 保存初始状态
	if err := h.saveTaskResult(ctx, taskResult); err != nil {
		httpResponse.Message = fmt.Sprintf("保存任务状态失败: %v", err)
		return httpResponse, fmt.Errorf(httpResponse.Message)
	}

	// 5. 启动工作池处理任务
	h.processTask(ctx, task, taskResult)

	httpResponse.Code = http.StatusOK
	return httpResponse, nil
}

// processTask 异步处理任务
func (h *TaskHandler) processTask(ctx context.Context, task *models.Task, taskResult *models.TaskResult) {

	// 创建工作池
	workers := h.getWorkerCount(task.MaxWorkers)
	itemChan := make(chan map[string]interface{}, len(task.Items))
	resultChan := make(chan models.ItemTaskResult, len(task.Items))

	// 启动工作协程
	var wg sync.WaitGroup
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go h.worker(ctx, &wg, itemChan, resultChan, task)
	}

	// 分发任务项
	go func() {
		for _, item := range task.Items {
			select {
			case <-ctx.Done():
				return
			case itemChan <- item:
			}
		}
		close(itemChan)
	}()

	// 等待所有工作协程完成
	go func() {
		wg.Wait()
		close(resultChan)
	}()

	// 处理结果
	h.handleResults(ctx, resultChan, taskResult)
}

// worker 工作协程
func (h *TaskHandler) worker(
	ctx context.Context,
	wg *sync.WaitGroup,
	itemChan <-chan map[string]interface{},
	resultChan chan<- models.ItemTaskResult,
	task *models.Task,
) {
	defer wg.Done()

	p := h.processors[task.Type]

	for item := range itemChan {
		select {
		case <-ctx.Done():
			return
		default:
			// 处理单个任务项

			result := p.Process(ctx, item, task.CommonParams)

			select {
			case <-ctx.Done():
				return
			case resultChan <- result:
			}
		}
	}
}

// handleResults 处理结果
func (h *TaskHandler) handleResults(
	ctx context.Context,
	resultChan <-chan models.ItemTaskResult,
	taskResult *models.TaskResult,
) {
	for result := range resultChan {
		// 更新结果
		taskResult.Results[result.ItemID] = result

		// 更新统计
		if result.Error != nil {
			taskResult.Failed++
		} else {
			taskResult.Success++
		}

		// 中断任务
		if result.IsBreak {
			break
		}

		// 保存中间状态
		if err := h.saveTaskResult(ctx, taskResult); err != nil {
			log.Printf("保存任务状态失败: %v", err)
		}
	}

	// 更新最终状态
	if taskResult.Status != "failed" {
		if taskResult.Failed > 0 {
			taskResult.Status = "completed_with_errors"
		} else {
			taskResult.Status = "completed"
		}
	}
	taskResult.EndTime = time.Now().Unix()

	// 保存最终结果
	if err := h.saveTaskResult(ctx, taskResult); err != nil {
		log.Printf("保存最终结果失败: %v", err)
	}
}

// saveTaskResult 保存任务结果到Redis
func (h *TaskHandler) saveTaskResult(ctx context.Context, result *models.TaskResult) error {
	data, err := json.Marshal(result)
	if err != nil {
		return err
	}

	key := h.config.Redis.GetKey(fmt.Sprintf("result:%s", result.ID))
	return h.rdb.Set(ctx, key, string(data), time.Duration(h.config.Task.ResultTTL)*time.Second).Err()
}

func (h *TaskHandler) ReadResult(ctx context.Context, taskID string) (models.TaskResult, error) {
	key := h.config.Redis.GetKey(fmt.Sprintf("result:%s", taskID))
	fmt.Println(key)
	val, err := h.rdb.Get(ctx, key).Bytes()
	if err != nil {
		return models.TaskResult{}, err
	}
	data := models.TaskResult{}
	err = json.Unmarshal(val, &data)
	if err != nil {
		return models.TaskResult{}, err
	}

	return data, nil
}
